-module(ezlinker_core_plugin).

-include("ezlinker_core_plugin.hrl").
-include_lib("emqx/include/emqx.hrl").

-export([ register_metrics/0
        , load/0
        , unload/0
        ]).


-export([ on_message_publish/2]).

-define(LOG(Level, Format, Args), emqx_logger:Level("ezlinker_core_plugin: " ++ Format, Args)).

register_metrics() ->
    [emqx_metrics:new(MetricName) || MetricName <- ['ezlinker_core_plugin.message_publish']].

load() ->
    lists:foreach(
      fun({Hook, Fun, Filter}) ->
          load_(Hook, binary_to_atom(Fun, utf8), {Filter})
      end, parse_rule(application:get_env(?APP, hooks, []))).

unload() ->
    lists:foreach(
      fun({Hook, Fun, _Filter}) ->
          unload_(Hook, binary_to_atom(Fun, utf8))
      end, parse_rule(application:get_env(?APP, hooks, []))).


load_(Hook, Fun, Params) ->
        case Hook of
            'message.publish'     -> emqx:hook(Hook, fun ?MODULE:Fun/2, [Params])
        end.

unload_(Hook, Fun) ->
        case Hook of
            'message.publish'     -> emqx:unhook(Hook, fun ?MODULE:Fun/2)
        end.

%%--------------------------------------------------------------------
%% Message publish
%%--------------------------------------------------------------------
on_message_publish(Message = #message{topic = <<"$SYS/", _/binary>>}, _Env) ->
    {ok, Message};
on_message_publish(Message = #message{ topic = Topic }, {Filter}) ->

    EZlinkerNodeName = application:get_env(?APP, node_name, 'ezlinker1@127.0.0.1'),
    with_filter(
      fun() ->
        emqx_metrics:inc('ezlinker_core_plugin.message_publish'),
        { ezlinker_node1_receive_loop , list_to_atom(EZlinkerNodeName) } ! { node() , Message },
        {ok, Message}
      end, Message, Topic, Filter).



%%--------------------------------------------------------------------
%% Internal functions
%%--------------------------------------------------------------------
parse_rule(Rules) ->
    parse_rule(Rules, []).
parse_rule([], Acc) ->
    lists:reverse(Acc);
parse_rule([{Rule, Conf} | Rules], Acc) ->
    Params = emqx_json:decode(iolist_to_binary(Conf)),
    Action = proplists:get_value(<<"action">>, Params),
    Filter = proplists:get_value(<<"topic">>, Params),
    parse_rule(Rules, [{list_to_atom(Rule), Action, Filter} | Acc]).

with_filter(Fun, _, _, undefined) ->
    Fun();
with_filter(Fun, Msg, Topic, Filter) ->
    case emqx_topic:match(Topic, Filter) of
        true  -> Fun();
        false -> {ok, Msg}
    end.

